package com.kafka.sync1;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;

import java.util.Properties;

/**
 * @author xq
 * @Description: 定义解析命令行
 * @date 2020/6/1115:01
 */
public class CliDefinition {

    private static Properties config = null;

    public static synchronized Properties getConfig(String[] args) {
        if (config == null){
            final Options options = new Options();
            Option kc = new Option("kc", "krb5Conf", true, "krb5Conf");
            Option jc = new Option("jc", "jaasConf", true, "jaasConf");
            Option bu = new Option("bu", "brokeUrl", true, "kafka.brokeUrl");
            Option to = new Option("to", "topic", true, "kafka.topic");
            Option cg = new Option("cg", "consumerGroup", true, "kafka.consumerGroup");
            Option sk = new Option("sk", "key", true, "source.key");
            Option tt = new Option("tt", "table", true, "target.table");
            Option ju = new Option("ju", "table", true, "target.jdbc.url");
            Option jn = new Option("jn", "table", true, "target.jdbc.username");
            Option jp = new Option("jp", "table", true, "target.jdbc.password");
            Option rc = new Option("rc", "table", true, "db.max.retryCount");
            Option fc = new Option("fc", "table", true, "db.max.failCount");
            options.addOption(kc).addOption(jc).addOption(bu).addOption(to);
            options.addOption(cg).addOption(sk).addOption(tt).addOption(ju);
            options.addOption(jn).addOption(jp).addOption(rc).addOption(fc);

            CommandLineParser parser = new DefaultParser();
            CommandLine cl;
            try {
                cl = parser.parse(options, args);
            } catch (ParseException e) {
                throw new RuntimeException("命令行解析失败");
            }
            config = new Properties();
            config.put(Config.KRB5_CONF.value(), cl.getOptionValue("kc"));
            config.put(Config.JAAS_CONF.value(), cl.getOptionValue("jc"));
            config.put(Config.KAFKA_BROKE_URL.value(), cl.getOptionValue("bu"));
            config.put(Config.KAFKA_TOPIC.value(), cl.getOptionValue("to"));
            config.put(Config.KAFKA_CONSUMER_GROUP.value(), cl.getOptionValue("cg"));
            config.put(Config.SOURCE_KEY.value(), cl.getOptionValue("sk"));
            config.put(Config.TARGET_TABLE.value(), cl.getOptionValue("tt"));
            config.put(Config.TARGET_JDBC_URL.value(), cl.getOptionValue("ju"));
            config.put(Config.TARGET_JDBC_USERNAME.value(), cl.getOptionValue("jn"));
            config.put(Config.TARGET_JDBC_PASSWORD.value(), cl.getOptionValue("jp"));
            if(cl.getOptionValue("rc") != null){
                config.put(Config.DB_MAX_RETRY_COUNT.value(), cl.getOptionValue("rc"));
            }else{
                config.put(Config.DB_MAX_RETRY_COUNT.value(), "0");
            }
            if (cl.getOptionValue("fc") != null){
                config.put(Config.DB_MAX_FALI_COUNT.value(), cl.getOptionValue("fc"));
            }else {
                config.put(Config.DB_MAX_FALI_COUNT.value(), "5");
            }

        }
        return config;
    }

    public static Properties getConfig(){
        return config;
    }

}
